package preprocess;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;

import trouger.dmt.JoinTableRunner;
import trouger.dmt.JoinType;
import utils.Utils;
import extractor.UserExtendedExtractor;

public class ACookieUserIdMapFilter implements Tool{
	public static String getACookieUserIdMapFilteredPath(String date) {
		return Utils.rootDir + date + "/ACookieUserIdMapFiltered/";
	}
	@Override 
	public Configuration getConf() {
		return null;
	}
	
	@Override
	public void setConf(Configuration conf) {
		
	}
	/*
	 * (non-Javadoc)
	 * UserFiltered Table
	 * Path: /group/tbsc-dev/xzwang/clickprediction/$date/ACookieUserIdMapFiltered/
	 * 1.ACookie
	 * 2.UserId 数字
	 * 3.UserId 32位
	 * */	 
	@Override
	public int run(String[] args) throws Exception {
		String aCookieUserIdMapFile = MappingACookieUserId.getACookieUserIdMapPath(args[0]);
		String output = getACookieUserIdMapFilteredPath(args[0]);
		String userExtenedProfile = UserExtendedExtractor.getUserExtendedProfilePath(args[0]);
		System.out.println("start job FilterAcookieUserIdMap...");
		System.out.println("userExtenedProfile: " + userExtenedProfile);
		System.out.println("aCookieUserIdMapFile: " + aCookieUserIdMapFile);
		System.out.println("output: " + output);
		if (Utils.exitSystem(args))
			return -1;
		JoinTableRunner r = new JoinTableRunner();
		r.addSourceTable(aCookieUserIdMapFile, "UTF-8", "\t");
		r.addSourceTable(userExtenedProfile, "UTF-8", "\t");
		r.addKeyGroup("1.2, 2.1");	
		r.setJoinType(JoinType.InnerJoin);
		r.setResultFieldsEx("1.1, 1.2, 2.3");
		r.setResultTable(output);
		r.run("FilterAcookieUserIdMap");
		return 0;		 	
	}
	
	/**
	 * @param args
	 */
	public static void main(String[] args) throws Exception {
		Utils.checkArgs(args);
		
		// TODO Auto-generated method stub
		int res = ToolRunner.run(new ACookieUserIdMapFilter(), args);
		System.exit(res);
	}
}
